package com.kafka.kafkatest.controller;

import cn.hutool.json.JSONObject;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.kafka.kafkatest.entity.AigcIntoEntity;
import com.kafka.kafkatest.entity.MsgLog;
import com.kafka.kafkatest.server.WebSocketServer;
import com.kafka.kafkatest.service.DataProcessingService;
import com.kafka.kafkatest.service.MsgLogService;
import com.kafka.kafkatest.util.KafkaProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executor;

@RestController
@RequestMapping("/test")
@Slf4j
public class TestController {


    @Resource
    private KafkaProducer kafkaProducer;

    @Autowired
    private WebSocketServer webSocketServer;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private MsgLogService msgLogService;

    @Value("${aigcHandleCount}")
    private Integer aigcHandleCount;

    @GetMapping("/send1")
    public String testSend() throws InterruptedException {

        ValueOperations operations = redisTemplate.opsForValue();
        Long num = operations.increment("num",1);
        if ( num>4){
            //发送客户消息正在排队
            JSONObject msg=new JSONObject();
            msg.set("code","0");
            msg.set("msg","正在排队中");
            webSocketServer.sendInfo(msg.toString(),"11");
        }
        processData("11111111111");
        return num.toString() ;
    }



    @PostMapping("/send2")
    public String testSend2() throws InterruptedException {

        ValueOperations operations = redisTemplate.opsForValue();
        try {
            MsgLog msgLog=new MsgLog();
            msgLog.setMsgId(IdWorker.getIdStr());
            msgLog.setTopicName("aigc");
            msgLogService.getOne(new LambdaQueryWrapper<MsgLog>().eq(MsgLog::getMsgId,"234324"));
            msgLogService.save(msgLog);
            Long num = operations.increment("num",1);
            System.out.println("当前并发num的值");
            if ( num>(aigcHandleCount-1)){
                //如果超过指定并发数量则丢到消息队列处理
                AigcIntoEntity entity=new AigcIntoEntity();
                entity.setFileName("11111111111111111");
                entity.setFileUrl("222222222222222");
                entity.setHeight("100");
                entity.setWidth("200");
                kafkaProducer.sendAigcMsg(JSON.toJSONString(entity));
//                kafkaProducer.sendAigcMsg("111111111111111111111");
                //发送客户消息正在排队
                return "目前比较繁忙,处理完后会保存在生成的相册中！";
            }
            Thread.sleep(4000);
            log.info("生成aigc成功！！！！！！！！！！");
            System.out.println("当前并发次数"+num);
            msgLog.setTopicName("aigc1");
            msgLogService.updateById(msgLog);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            //释放可以处理的的并发
            operations.decrement("num",1);
        }
        return "ok";
    }


    @Autowired
    private DataProcessingService dataProcessingService;

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;



    public String processData(String data) {
        dataProcessingService.processData(data);
        return "Data processing request received";
    }

}
